package com.shujia.flink.window

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{GlobalWindows, ProcessingTimeSessionWindows}
import org.apache.flink.streaming.api.windowing.time.Time

object Demo2SessionWindow {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val ds = env.socketTextStream("node1", 8888)


    ds
      .flatMap(_.split(","))
      .map(w => (w, 1))
      .keyBy(_._1)
      //当一定时间没有数据后触发窗口
      .window(ProcessingTimeSessionWindows.withGap(Time.seconds(3)))
      .reduce((x, y) => (x._1, x._2 + y._2))
      .print()


//    ds
//      .flatMap(_.split(","))
//      .map(w => (w, 1))
//      .keyBy(_._1)
//      .window(GlobalWindows.create())
//      .reduce((x, y) => (x._1, x._2 + y._2))
//      .print()

    env.execute()


  }
}
